#include <assert.h>
#include <unistd.h>
#include <pthread.h>

thread_pool_t * creative_thread_pool (int max_thread_num)
{

	thread_pool *pool = NULL;
	
	assert(max_thread_num > 0);

	pool = (tread_pool_t *)malloc(sizeof(thread_pool_t));
	if (!pool) {
		return -1;
	}

	pool->threads = (pthread_t *)malloc(max_thread_num * sizeof(pthread_t));
	if (!pool->threads) {
		free(pool);
		return -1;
	}

	pthread_mutex_init(&(pool->queue_lock), NULL);
	pthread_cond_init(&(pool->queue_ready), NULL);

	pool->queue_head = NULL;

	pool->max_thread_num = max_thread_num;
	pool->cur_queue_size = pool->shutdown = 0;

	int i = 0;
	for (i = 0; i < max_thread_num; i++) {
		pthread_create(&(pool->threadid[i]), NULL, thread_routine, NULL);
	}
	return 0;
}


int pool_add_worker (thread_pool_t *pool, process_fun process, void *arg)
{
	thread_worker_t *worker = (thread_worker *)bcache_alloc(sizeof(thread_worker));
	if (worker == NULL) {
		return -1;
	}
	worker->process = process;
	worker->arg = arg;
	worker->next = NULL;


	/* lock worker queue */
	pthread_mutex_lock(&(pool->queue_lock));
	
	thread_worker *member = pool->queue_head;
	
	if (member != NULL) {
		while (member->next != NULL) {
			member = member->next;
		}
		member->next = newworker;
	} else {
		pool->queue_head = newworker;
	}

	assert (pool->queue_head != NULL);

	pool->cur_queue_size++;
	pthread_mutex_unlock(&(pool->queue_lock));
	pthread_cond_signal(&(pool->queue_ready));
	return 0;
}


int pool_destroy(struct mempool *pool)
{
	int i;
	
	if (pool->shutdown) {
		return -1;
	}
	pool->shutdown = 1;

	pthread_cond_broadcast(&(pool->queue_ready));

	for (i = 0; i < pool->max_thread_num; i++) {
		pthread_join(pool->threadid[i], NULL);
	}
	
	free (pool->threadid);

	thread_worker_t *head = NULL;
	while (pool->queue_head != NULL) {
		head = pool->queue_head;
		pool->queue_head = pool->queue_head->next;
		bcache_free(head);
	}

	pthread_mutex_destroy(&(pool->queue_lock));
	pthread_cond_destroy(&(pool->queue_ready));

	free(pool);

	return 0;
}


void * thread_routine (void *arg)
{
	printf ("starting thread 0x%x\n", pthread_self ());
	
	while (1) {

		pthread_mutex_lock(&(pool->queue_lock));
		//有可能惊群。多处理器里可能会同时唤醒多个线程
		while (pool->cur_queue_size == 0 
				&& !pool->shutdown) {
			printf("thread 0x%x is waiting\n", pthread_self ()); 
			pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));
		}

		if (pool->shutdown) {
			pthread_mutex_unlock(&(pool->queue_lock));
			printf("thread 0x%x will exit\n", pthread_self());
			pthread_exit(NULL);
		}

		printf("thread 0x%x is starting to work\n", pthread_self());

		assert(pool->cur_queue_size != 0);
		assert(pool->queue_head != NULL);

		pool->cur_queue_size--;
		thread_worker_t * worker = pool->queue_head;
		pool->queue_head = worker->next;
		pthread_mutex_unlock(&(pool->queue_lock));

		(*(worker->process))(worker->arg);
		bcache_free(worker);
		worker = NULL;
	}
	pthread_exit (NULL);
}
